package org.nbict.iot.task.transit;

import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.bolt.KafkaBolt;
import org.apache.storm.kafka.bolt.mapper.FieldNameBasedTupleToKafkaMapper;
import org.apache.storm.kafka.bolt.selector.FieldNameTopicSelector;
import org.apache.storm.kafka.spout.KafkaSpout;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.topology.TopologyBuilder;

import java.util.Properties;
import java.util.regex.Pattern;

/**
 * 报文解析拓扑
 * Created by songseven on 18/6/19.
 */
@Deprecated
public class ProtocolParserTopology {

    public static void main(String[] args) throws InvalidTopologyException,
            AuthorizationException, AlreadyAliveException {

        TopologyBuilder builder = new TopologyBuilder();

        KafkaSpoutConfig.Builder<String, byte[]> kafkaBuilder = new
                KafkaSpoutConfig.Builder("192.168.1.68:9092", StringDeserializer //192.168.1.68:9092
                .class, ByteArrayDeserializer.class, Pattern.compile("raw-.*-simple")); //需要监听多个topic 需要测试 raw-.*-simple

        kafkaBuilder.setGroupId("iot-car-raw-parser-group-1");
        KafkaSpoutConfig build = kafkaBuilder.build();
        KafkaSpout kafkaSpout = new KafkaSpout(build);

        //step1. 报文读取流
        builder.setSpout("kafkaSpout", kafkaSpout, 1);

        //step2. 报文解析流
        builder.setBolt("parserBolt", new BinaryParserBolt(), 3)
                .shuffleGrouping("kafkaSpout");

        //step3. 设置输出到kafka
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.1.68:9092"); //192.168.1.83:9092
        props.put("acks", "1");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        KafkaBolt outBolt = new KafkaBolt()
                .withProducerProperties(props)
                .withTopicSelector(new FieldNameTopicSelector("platform_id", "transit")) //default topic name
                .withTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper(null, "parsed"));
        builder.setBolt("forward2Kafka", outBolt, 2).shuffleGrouping("parserBolt", "correct");

        Config config = new Config();
        config.put("kafka.topic.wildcard.match", true);

        if (args.length > 0) {
            //集群提交模式
            config.setDebug(false);
            StormSubmitter.submitTopology(args[0], config, builder
                    .createTopology());
        } else {
            //本地测试模式
            config.setDebug(false);
            //设置1个进程
            config.setNumWorkers(1);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("iot-car-parser", config, builder
                    .createTopology());
        }
    }
}
